package Test;

import java.util.regex.*;
import java.io.IOException;
import java.util.*;
import java.util.Date;
import java.text.DateFormat;
import java.text.SimpleDateFormat;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

  public class Test {

   public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, LongWritable> {
    private Text word = new Text(); //session
    public void map(LongWritable key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter) throws IOException {
      String line = value.toString();
      
      word.set(str);
      output.collect(word, new LongWritable(1));
    }
   }

  public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {
    //ArrayList list = new ArrayList();
    public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
      String st = "";
      String[] ary;
      String concate = "";
      String[] items;
      String sum="";
      
      Vector v = new Vector();
      String value = "";
      String sid = "";
      String rest = "";
      String result = "";
      String[] tokens;
      String token = "";
      Text text = new Text();
      // while(values.hasNext()){
        // output.collect(key, new Text(values.next()+""));
      // }
      //v.clear();
      
      // result="";
      // int x = 1;
      // while(values.hasNext()){
        // token = values.next()+"";
        //result+="["+token+"-"+String.valueOf(x)+"];";
        // output.collect(key, new Text(result));
        // x ++;
      // }
      
      int x = 1;
      while(values.hasNext()){
        token = values.next()+"";
        if (token.length() == 3) {
          sid = token;
          //if ((!sid.equals("")) && (sid.indexOf("<") > 0)) {
          DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
          Date date = new Date();
          String time =  dateFormat.format(date);
          result = "<"+sid+"-"+String.valueOf(x)+"-"+time+">";
          if ((text.toString()).equals("")){
            text.set(result);
            output.collect(key, text);          
          }
        }
        else {
          rest = token;
          //if ((!rest.equals("")) && (rest.indexOf("[") > 0)) {
          DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
          Date date = new Date();
          String time =  dateFormat.format(date);
          result = "["+rest+"-"+String.valueOf(x)+"-"+time+"]";
          
          if ((text.toString()).equals("")){
            text.set(result);
            output.collect(key, text);          
          }
        }
        text.set("");
        x++;
      }
      if ((!sid.equals("")) && (!rest.equals(""))){
      }
      
      
      // int x = 1;
      // while(1==1){
        // try {
          // token = values.next()+"";
          // if (!token.equals("")) {
            // result+="["+token+"-"+String.valueOf(x)+"];";
          // }
        // }
        // catch(Exception e) {
          // break;
        // }
        // x++;
      // }
      
      
      // for(int i=0; i<v.size(); i++){
        // result += v.get(i);
      // }
      //output.collect(key, new Text(result));
      
      // Pattern ptn2 = Pattern.compile("([\\w\\d^\\t]+)");
      // Matcher mch2 = ptn2.matcher(result);
      // while (mch2.find()) {
        // sid = mch2.group(1);
      // }
      // output.collect(key, new Text(sid));
      //v.clear();
      // for(int i=0; i<v.size(); i++){
        // value = (String)v.get(i);
        // if (value.length() == 3) {
          // sid = value;
          // v.removeElementAt(i);
          // break;
        // }
      // }
      // for(int i=0; i<v.size(); i++){
        // value = (String)v.get(i);
        // result += "["+sid+"],<"+value+">\n";
      // }
      // output.collect(key, new Text(result));
      
      // DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss");
      // Date date = new Date();
      // String time =  dateFormat.format(date);
      // sum += "{" + time + "}"; 
      // output.collect(key, new Text(sum));
        // items = concate.split("\t");
        // for(int i=0; i<items.length; i++){
          // if (!items[i].equals("")) {
            // sum += "[" + sid + "]\t" + items[i] + "\n" + key + "\t";
          // }
         //sum += "[" + sid + "]\t" + (String)v.get(i) + "\n" + key + "\t";
         //sum += sid;
        //}
        //if (v.size() != 0) {
        // if (items.length != 0) {
          // sum = sum.substring(0,sum.length()-5);
        // }
      
      
      
      
      // while(values.hasNext()){
        // String st=values.next()+"";
        // if(st.length()==3)
          // sid=st;
        // else
          // list.add(st);
      // }
      
      // int i;
      // for(i=0; i<list.size()-1; i++){ //-1
       // sum += sid + "\t" + list.get(i) + "\n" + key + "\t";
      // }
      // if (list.size() != 0) {
        // sum += sid + "\t" + list.get(i);
      // }
      // output.collect(key, new Text(sum));
    }
  }

       public static void main(String[] args) throws Exception {
       JobConf conf = new JobConf(Test.class);
       conf.setJobName("Test");

       //conf.setNumReduceTasks(1);
       conf.setNumMapTasks(5);
       
       conf.setOutputKeyClass(Text.class);
       conf.setOutputValueClass(Text.class);

       conf.setMapperClass(Map.class);
       conf.setCombinerClass(Reduce.class);
       conf.setReducerClass(Reduce.class);

       conf.setInputFormat(TextInputFormat.class);
       conf.setOutputFormat(TextOutputFormat.class);

       FileInputFormat.setInputPaths(conf, new Path(args[0]));
       FileOutputFormat.setOutputPath(conf, new Path(args[1]));

       JobClient.runJob(conf);
     }
  }

